@on_task Decorator

The @on_task decorator provides a simple way to register functions as event-driven task handlers for agent deployment and background task processing.

Overview

The @on_task decorator automatically:

Registers your function as a task handler
Sets up event listening for incoming task execution requests
Handles both synchronous and asynchronous functions
Validates that the decorated function has a task parameter
Works with default configuration from environment variables

Syntax

@on_task
def your_task_handler(task):
    # Your task processing logic
    return task

Parameters

The decorator itself takes no parameters, but the decorated function must:

Accept a task parameter as its first argument
Return the modified task object

Usage Examples

Basic Synchronous Handler

from xpander_sdk import on_task

@on_task
def handle_task(task):
    print(f"Processing task: {task.id}")
    print(f"Task input: {task.input.text}")
    
    # Your custom logic here
    # ... processing logic ...
    
    # Optional: Set internal status once before returning/saving
    task.internal_status = "Custom processing completed successfully"
    task.result = "Task processed successfully"
    
    return task  # internal_status is saved when task is returned

Asynchronous Handler

from xpander_sdk import on_task
import asyncio

@on_task
async def handle_task_async(task):
    print(f"Processing task: {task.id}")
    
    # Simulate async work
    await asyncio.sleep(1)
    
    # Your custom async logic here
    task.result = "Async task processed successfully"
    
    # Optional: Set internal status once before returning (max 255 characters)
    task.internal_status = "Async processing completed"
    
    return task  # internal_status is persisted when task is returned

Task Processing with Error Handling

from xpander_sdk import on_task
from xpander_sdk.modules.tasks.models.task import AgentExecutionStatus

@on_task
def handle_task_with_error_handling(task):
    try:
        print(f"Processing task: {task.id}")
        
        # Your processing logic
        if task.input.text == "error":
            raise ValueError("Simulated error")
            
        task.result = "Task completed successfully"
        task.status = AgentExecutionStatus.Completed
        
    except Exception as e:
        print(f"Task failed: {str(e)}")
        task.result = f"Task failed: {str(e)}"
        task.status = AgentExecutionStatus.Failed
    
    return task

Processing Different Input Types

from xpander_sdk import on_task

@on_task
def handle_multimodal_task(task):
    print(f"Processing task: {task.id}")
    
    processed_items = []
    
    # Handle text input
    if task.input.text:
        print(f"Text input: {task.input.text}")
        # Process text
        processed_items.append("text")
    
    # Handle file inputs
    if task.input.files:
        print(f"Files to process: {len(task.input.files)}")
        for file_url in task.input.files:
            print(f"Processing file: {file_url}")
            # Process each file
            processed_items.append("file")
    
    task.result = "Multimodal processing completed"
    
    # Optional: Set internal status once with summary before returning
    task.internal_status = f"Processed {len(processed_items)} items: {', '.join(processed_items)}"
    
    return task  # internal_status is saved when task is returned

Using Optional Internal Status for Context Information

from xpander_sdk import on_task

@on_task
async def handle_task_with_context(task):
    """Example showing optional internal_status for context information."""
    print(f"Processing task: {task.id}")
    
    # internal_status is completely optional - you can omit it entirely
    # When used, it's limited to 255 characters and set once before saving
    
    try:
        # Validation phase
        if not task.input.text:
            task.result = "No input text provided"
            # Optional: Set internal status once before returning
            task.internal_status = "Validation failed: no input provided"
            return task
        
        # Main processing logic
        word_count = len(task.input.text.split())
        result = f"Processed text with {word_count} words"
        
        # Set final result
        task.result = result
        
        # Optional: Set internal status once with additional context
        task.internal_status = f"Text analysis completed: {word_count} words processed"
        
    except Exception as e:
        task.result = f"Processing failed: {str(e)}"
        # Optional: Set internal status once for error context
        task.internal_status = f"Error during text processing: {str(e)[:200]}"
    
    return task  # internal_status is persisted when task is returned

# Note: You can also ignore internal_status completely - it's optional!
@on_task
def simple_handler_without_internal_status(task):
    """Simple handler that doesn't use internal_status at all."""
    task.result = "Processed without internal status tracking"
    return task  # No internal_status - perfectly valid!

Using Internal Status with Explicit Saving

from xpander_sdk import on_task

@on_task
async def handle_task_with_saving(task):
    """Example showing internal_status with explicit task saving."""
    
    # Process the task
    task.result = "Processing completed"
    
    # Optional: Set internal status once before saving
    task.internal_status = "Custom processing pipeline completed"
    
    # Save the task with the internal status
    await task.asave()  # internal_status is persisted here
    
    return task

Behavior

Automatic Event Listening

When you use the @on_task decorator, it automatically:

Sets up event listening for your agent
Waits for incoming task execution requests
Calls your decorated function when tasks arrive
Handles the task lifecycle (status updates, result saving)

Function Requirements

Your decorated function must:

Accept a task parameter
Return the modified task object
Handle any exceptions internally (recommended)

Configuration

The decorator uses:

Environment variables for SDK configuration (XPANDER_API_KEY, XPANDER_ORGANIZATION_ID, etc.)
Default agent ID from XPANDER_AGENT_ID environment variable

Task Object Properties

The task parameter passed to your function contains:

# Task identification
task.id                    # str: Unique task identifier
task.agent_id             # str: Associated agent ID
task.organization_id      # str: Organization identifier

# Task input
task.input.text           # str: Text prompt/instruction
task.input.files          # List[str]: URLs of input files
task.input.user           # User: User details (if provided)

# Task status and results
task.status               # AgentExecutionStatus: Current status
task.internal_status      # Optional[str]: Custom status (max 255 chars, optional)
task.result               # str: Task result (set by your handler)

# Timestamps
task.created_at           # datetime: When task was created
task.started_at           # datetime: When execution started
task.finished_at          # datetime: When execution finished

# Configuration
task.output_format        # OutputFormat: Desired output format
task.output_schema        # Dict: JSON schema for output validation
task.events_streaming     # bool: Whether event streaming is enabled

Best Practices

1. Always Return the Task

@on_task
def good_handler(task):
    # Process task
    task.result = "Done"
    return task  # ✅ Always return

2. Handle Errors Gracefully

@on_task
def robust_handler(task):
    try:
        # Your logic here
        task.result = "Success"
    except Exception as e:
        task.result = f"Error: {str(e)}"
        task.status = AgentExecutionStatus.Failed
    return task

3. Set Appropriate Status

@on_task
def status_aware_handler(task):
    # Process task
    if success:
        task.status = AgentExecutionStatus.Completed
    else:
        task.status = AgentExecutionStatus.Failed
    return task

4. Validate Input

@on_task
def validating_handler(task):
    if not task.input.text and not task.input.files:
        task.result = "Error: No input provided"
        task.status = AgentExecutionStatus.Failed
        return task
    
    # Process valid input
    task.result = "Processing completed"
    return task

5. Use Internal Status Optionally for Context Information

@on_task
def context_tracking_handler(task):
    # Validation logic
    if not task.input.text:
        task.result = "Error: No input provided"
        task.status = AgentExecutionStatus.Failed
        # Optional: Set internal status once before returning (max 255 chars)
        task.internal_status = "Validation failed: no input provided"
        return task
    
    # ... processing logic ...
    processed_items = len(task.input.text.split())
    task.result = "Processing completed"
    
    # Optional: Set internal status once with additional context before returning
    task.internal_status = f"Successfully processed {processed_items} items"
    return task  # internal_status is saved when task is returned

# Or completely omit internal_status - it's optional!
@on_task
def minimal_handler(task):
    task.result = "Processed"
    return task  # No internal_status needed - perfectly valid!

[AgentExecutionStatus](/API reference/types#agentexecutionstatus): Task execution statuses
[AgentExecutionInput](/API reference/types#agentexecutioninput): Task input structure
[OutputFormat](/API reference/types#outputformat): Output format options

[Events Module](/API reference/events): Full events module documentation
[Tasks Module](/API reference/tasks): Task management and execution
[Agents Module](/API reference/agents): Agent creation and management

Overview

Core Modules

API Reference

CLI Reference

@on_task Decorator API Reference

@on_task Decorator

Overview

Syntax

Parameters

Usage Examples

Basic Synchronous Handler

Asynchronous Handler

Task Processing with Error Handling

Processing Different Input Types

Using Optional Internal Status for Context Information

Using Internal Status with Explicit Saving

Behavior

Automatic Event Listening

Function Requirements

Configuration

Task Object Properties

Best Practices

1. Always Return the Task

2. Handle Errors Gracefully

3. Set Appropriate Status

4. Validate Input

5. Use Internal Status Optionally for Context Information

Overview

Core Modules

API Reference

CLI Reference

​@on_task Decorator

​Overview

​Syntax

​Parameters

​Usage Examples

​Basic Synchronous Handler

​Asynchronous Handler

​Task Processing with Error Handling

​Processing Different Input Types

​Using Optional Internal Status for Context Information

​Using Internal Status with Explicit Saving

​Behavior

​Automatic Event Listening

​Function Requirements

​Configuration

​Task Object Properties

​Best Practices

​1. Always Return the Task

​2. Handle Errors Gracefully

​3. Set Appropriate Status

​4. Validate Input

​5. Use Internal Status Optionally for Context Information

​Related Types

​Related Documentation

@on_task Decorator

Overview

Syntax

Parameters

Usage Examples

Basic Synchronous Handler

Asynchronous Handler

Task Processing with Error Handling

Processing Different Input Types

Using Optional Internal Status for Context Information

Using Internal Status with Explicit Saving

Behavior

Automatic Event Listening

Function Requirements

Configuration

Task Object Properties

Best Practices

1. Always Return the Task

2. Handle Errors Gracefully

3. Set Appropriate Status

4. Validate Input

5. Use Internal Status Optionally for Context Information

Related Types

Related Documentation